use std::fs::File;
use std::io::Read;
use std::io::BufRead;
use std::io::Write;
use std::path::Path;
use failure::Error;
use std::convert::TryInto;
use prost::Message;
use rayon::prelude::*;

pub mod proto {
    include!(concat!(env!("OUT_DIR"), "/coin.proto.rs"));
}

// /remote/iosg/strat-1/buckets/log.raw.coin/live/og_proto_log/strategy-05.eu-central-1/20190708                <= server name / date
// unhedged_mm_3.py.xunke01.Spot.Bitstamp.20190708-010018.1562547618748907000.gz       <=  strategy.account.*   

// e.exchange_type
// e.symbol

const MAGIC:&[u8] = b"\x00\x31\x41\x59\x00\x01\x00\x00";
type Order = proto::OrderGatewayLog;

fn read_order_log_sub<R:Read>(mut f: R) -> Result<Vec<Order>, Error> {
    let mut buf = vec![0u8; 4096*4];
    let mut ret = Vec::new();

    f.read_exact(&mut buf[..MAGIC.len()])?;
    if MAGIC != &buf[0..MAGIC.len()] {
        panic!("MAGIC mismatch");
    }

    loop {
        if f.read_exact(&mut buf[..8]).is_err() {
            break;
        }

        let length = i64::from_le_bytes((&buf[..8]).try_into()?) as usize;

        // f.seek(SeekFrom::Current(4+32))?;
        f.read_exact(&mut buf[..4+32])?;
        f.read_exact(&mut buf[..(length-32)])?;
        
        let lg = proto::OrderGatewayLog::decode(&buf[..(length-32)])?;
        f.read_exact(&mut buf[..4])?;
        if proto::order_gateway_log::LogType::from_i32(lg.r#type.unwrap()).unwrap() == proto::order_gateway_log::LogType::OrderEvent {
            // match proto::order_event::OrderEventType::from_i32(lg.event.as_ref().unwrap().r#type.unwrap()).unwrap() {
            //     proto::order_event::OrderEventType::OrderSubmitted => {
            //         cnt += 1;
            //     }
            //     proto::order_event::OrderEventType::OrderAccepted => {
                    
            //     }
            //     _ => {
            //     }
            // }
            if let Some(s) = filter.symbol {
                let symbol = lg.event.as_ref().unwrap().symbol;
                if s != symbol {
                    continue;
                }
            }
            ret.push(lg);
        }
    }
    Ok(ret)
}

fn read_order_log<P:AsRef<Path>>(path: P) -> Result<Vec<Order>, Error> {
    let path = path.as_ref();
    let f = File::open(path)?;

    let ret = if path.ends_with(".gz") {
        let f = flate2::read::GzDecoder::new(f);
        read_order_log_sub(f)?
    } else {
        read_order_log_sub(f)?
    };
    
    Ok(ret)
}

fn read_order_logs(filter: Filter, out: &str) -> Result<(), Error> {
    /*
        .
        ./strategy-11.eu-west-1
        ./strategy-11.eu-west-1/20180531
        ./strategy-11.eu-west-1/20180531/pass_unhedge_2.py.Bitmex.Futures.20180531-112038
        ./strategy-11.eu-west-1/20180531/pass_unhedge_2.py.Bitmex.Futures.20180531-132055
                                               0        1    2       3      4
        ./strategy-11.eu-west-1/20180531/pass_unhedge_2.py.Bitmex.Futures.20180531-125053
        0  1                     2              3        
    */

    let f = File::open("ls_og_proto_logs.txt")?;
    let f = std::io::BufReader::new(f);
    let paths:Vec<_> = f.lines().map(|line| line.unwrap()).filter(|line| {
        // 3 slashes
        let dirs:Vec<_> = line.split('/').collect();
        match dirs.len() {
            l if l < 4 => {return false;}
            l if l > 4 => {
                // unexpected
                println!("more than 3 slashes : {}", line);
                return false;
            }
            _ => {
                let date = dirs[2];

                if let Some(d) = filter.date {
                    if d != date {
                        return false;
                    }
                }

                let tokens:Vec<&str> = dirs[3].split('.').collect();
                let gz_one = if tokens.last()==Some(&"gz") { 1 } else { 0 };
                if tokens.len() < 5+gz_one || tokens.len() > 6+gz_one {
                    return false;
                }
                // 0               1  2       3             4          5
                // <strategy_name>.py.<owner>.<market_type>.<exchange>.<YYmmDD-HHMMSS>(.<timestamp_nano_sec>)      6
                // test_mm_ma_strategy2.py.jshin0.Futures.Bitmex.20180611-084408.1528706648659898000              .gz

                let strategy = tokens[0];
                let owner = tokens[2];
                let market_type = tokens[3];
                let exchange = tokens[4];
                // let _datetime = tokens[5];

                if let Some(s) = filter.strategy {
                    if s != strategy {
                        return false;
                    }
                }
                if let Some(o) = filter.owner {
                    if o != owner {
                        return false;
                    }
                }
                if let Some(m) = filter.market_type {
                    if m != market_type {
                        return false;
                    }
                }
                if let Some(e) = filter.exchange {
                    if e != exchange {
                        return false;
                    }
                }
                return true;
            }
        }
    }).collect();
    
    
    let pb = indicatif::ProgressBar::new(paths.len() as u64);
    // paths.iter().for_each(|p| {
    let vecs:Vec<_> = paths.par_iter().map(|p| {
        pb.inc(1);
        let p = format!("/remote/iosg/strat-1/buckets/log.raw.coin/live/og_proto_log/{}", p);
        read_order_log(p).unwrap()
    }).collect();
    let mut vec:Vec<Order> = itertools::concat(vecs);
    pb.finish_with_message("done");

    println!("sorting {} orders by time..", vec.len());
    vec.sort_by(|a, b| {
        a.event.as_ref().unwrap().event_time.partial_cmp(&b.event.as_ref().unwrap().event_time).unwrap()
    });

    println!("saving..");
    
    let f = File::create(out)?;
    let mut f = std::io::BufWriter::new(f);
    let mut buf = vec![];

    let pb = indicatif::ProgressBar::new(vec.len() as u64);
    for order in vec {
        f.write_all(&u64::to_le_bytes(order.encoded_len() as u64))?;

        buf.resize(order.encoded_len(), 0);
        order.encode(&mut buf)?;
        f.write_all(&buf)?;
        pb.inc(1);
    }
    pb.finish_with_message("done");

    Ok(())
}

#[derive(Debug)]
struct Filter<'a> {
    date: Option<&'a str>,
    strategy: Option<&'a str>,
    owner: Option<&'a str>,
    market_type: Option<&'a str>,
    exchange: Option<&'a str>,
    symbol: Option<&'a str>,
}

fn main() {
    use clap::{App, Arg};
    let matches = App::new("order-converter")
        .version("1.0")
        .author("Donggu Kang <donggu@prestolabs.com>")
        .about("Extract order with a given strategy, exchange, date")
        .arg(Arg::with_name("date")
            .short("d")
            .long("date")
            .takes_value(true))
        .arg(Arg::with_name("strategy")
            .short("s")
            .long("strategy")
            .takes_value(true))
        .arg(Arg::with_name("owner")
            .long("owner")
            .takes_value(true))
        .arg(Arg::with_name("market_type")
            .short("m")
            .long("market_type")
            .takes_value(true))
        .arg(Arg::with_name("exchange")
            .short("e")
            .long("exchange")
            .takes_value(true))
        .arg(Arg::with_name("symbol")
            .short("s")
            .long("symbol")
            .takes_value(true))
        .arg(Arg::with_name("out")
            .long("out")
            .takes_value(true)
            .required(true))
        .get_matches();

    // println!("{}", std::mem::size_of::<proto::OrderGatewayLog>()); // 1800 bytes
    
    let filter = Filter {
        date: matches.value_of("date").clone(),
        strategy: matches.value_of("strategy").clone(),
        owner: matches.value_of("owner").clone(),
        market_type: matches.value_of("market_type").clone(),
        exchange: matches.value_of("exchange").clone(),
        symbol: matches.value_of("symbol").clone(),
    };

    println!("{:?}", filter);
    read_order_logs(filter, matches.value_of("out").unwrap()).unwrap();
}
